package com.gourd.common.kafka.service;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;
import org.springframework.util.concurrent.ListenableFuture;
 
/**
 * @author gourd
 * @create 2018/11/1
 */
@Service
@Slf4j
public class KafkaService {
 
    private final KafkaTemplate<String, String> kafkaTemplate;
   
 
    @Autowired
    public KafkaService(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = (KafkaTemplate<String, String>) kafkaTemplate;
    }

    /**
     * 发送消息
     * @param topic
     * @param message
     */
    public void send(String topic, String message) {
        ListenableFuture<SendResult<String, String>> resultListenableFuture = kafkaTemplate.send(topic, message);
        resultListenableFuture.addCallback(
                successCallback -> log.info("^o^= 发送成功：topic= " + topic + " message= " + message),
                failureCallback -> log.info("^o^= 发送失败：topic= " + topic + " message= " + message));
    }
 

}